package com.hulk.rpc.transport.server.rest;

import com.hulk.rpc.config.common.HostPort;
import com.hulk.rpc.filter.RestServerFilter;
import com.hulk.rpc.invoke.ServiceInvokeFactory;
import com.hulk.rpc.serialize.impl.json.JsonMapper;
import com.hulk.rpc.transport.server.Server;
import com.hulk.rpc.transport.server.rest.handler.NettyRestChannelInitializer;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.WriteBufferWaterMark;
import io.netty.channel.epoll.EpollChannelOption;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.epoll.EpollServerSocketChannel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.concurrent.CopyOnWriteArrayList;

/**
 * @author xuxukang
 */
public class NettyRestServer implements Server {

  private final CopyOnWriteArrayList<RestServerFilter> serverFilters;
  private final HostPort hostPort;
  private final EventLoopGroup eventLoopGroup;
  private final ServiceInvokeFactory invokerFactory;
  private final JsonMapper jsonMapper;

  private volatile Channel channel;

  public NettyRestServer(CopyOnWriteArrayList<RestServerFilter> serverFilters, HostPort hostPort,
                         EventLoopGroup eventLoopGroup, ServiceInvokeFactory invokerFactory, JsonMapper jsonMapper) {
    this.serverFilters = serverFilters;
    this.hostPort = hostPort;
    this.eventLoopGroup = eventLoopGroup;
    this.invokerFactory = invokerFactory;
    this.jsonMapper = jsonMapper;
  }

  @Override
  public void start() throws InterruptedException {
    InetSocketAddress inet = new InetSocketAddress(hostPort.getHost(), hostPort.getPort());

    ServerBootstrap bootstrap = new ServerBootstrap();
    bootstrap.group(eventLoopGroup);

    bootstrap.option(ChannelOption.SO_BACKLOG, 1024);
    bootstrap.option(ChannelOption.SO_REUSEADDR, true);
    bootstrap.option(ChannelOption.SO_RCVBUF, 256 * 1024);

    if (eventLoopGroup instanceof EpollEventLoopGroup) {
      bootstrap.option(EpollChannelOption.SO_REUSEPORT, true);
      bootstrap.channel(EpollServerSocketChannel.class);
    } else if (eventLoopGroup instanceof NioEventLoopGroup) {
      bootstrap.channel(NioServerSocketChannel.class);
    }

    bootstrap.childHandler(new NettyRestChannelInitializer(invokerFactory, jsonMapper, serverFilters));

    bootstrap.childOption(ChannelOption.SO_REUSEADDR, true);
    bootstrap.childOption(ChannelOption.SO_RCVBUF, 256 * 1024);
    bootstrap.childOption(ChannelOption.SO_SNDBUF, 256 * 1024);
    bootstrap.childOption(ChannelOption.WRITE_BUFFER_WATER_MARK, //
            new WriteBufferWaterMark(1024 * 1024, 2048 * 1024));

    channel = bootstrap.bind(inet).sync().channel();

    System.out.println("NettyRestServer started. Listening on: " + hostPort);
  }

  @Override
  public void close() throws IOException {
    try {
      channel.closeFuture().sync();
    } catch (InterruptedException e) {
      throw new IOException(e);
    }
  }

}
